package com.chat.business.common.rabbitmq;

import com.chat.business.service.IHistoryMessageService;
import com.chat.common.entity.HistoryMessage;
import com.rabbitmq.client.Channel;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import java.io.IOException;
import java.util.function.Consumer;

/**
 * @author wei kunJie
 * @date 2024/4/22 18:15:57
 * <p>
 * 接收rabbitMQ消息
 */
@Configuration
@Slf4j
public class RabbitMqConsumer {
    @Resource
    private IHistoryMessageService historyMessageService;

    /**
     * 处理要保存的用户消息记录
     *
     * @return
     */
    @Bean
    Consumer<Message<HistoryMessage>> historyMessage() {
        return historyMessageMessage -> {
            //要保存到数据库的数据
            HistoryMessage historyMessage = historyMessageMessage.getPayload();
            log.info("持久化消息：{}", historyMessage);
            MessageHeaders headers = historyMessageMessage.getHeaders();
            //获取信道
            Channel channel = headers.get("amqp_channel", Channel.class);
            //rabbitmq消息id
            Long deliveryTag = headers.get("amqp_deliveryTag", Long.class);
            log.info("rabbitmq msg deliveryTag：{}", deliveryTag);
            try {
                assert channel != null : "频道 channel 为 bull";
                //保存到数据库
                historyMessageService.save(historyMessage);
                log.info("持久化消息完成{}", historyMessage);
                //手动ack  第二个参数为false是表示仅仅确认当前消息 true表示确认之前所有的消息
                //不ack是不会处理下一条消息的
                channel.basicAck(deliveryTag, false);
            } catch (IOException e) {
                e.printStackTrace();
                try {
                    //手动nack 告诉rabbitmq该消息消费失败  第三个参数：如果被拒绝的消息应该被重新请求，而不是被丢弃或变成死信，则为true
                    channel.basicNack(deliveryTag, false, true);
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        };
    }
}
